package com.index;

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

public class MergeIndex {

	public static final String INDEX_PATH = "/index/stable/";
	
	public static class Map extends MapReduceBase implements Mapper<LongWritable,Text,Text,Text> {
		//输入格式：tmp--> word|count@path|count
		//目标目录 1、为空
		//			2、和tmp一样
		//			3、word|count@path1|count&path2|count....
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] strs = line.split("@");
			if(strs.length == 2){
				String[] head = strs[0].split("|");
				String newKey = head[0];
				String newValue = head[1]+"@"+strs[1];
				output.collect(new Text(newKey),new Text(newValue));
			}
		}
	}
	
	
	public static class Reduce extends MapReduceBase implements Reducer<Text,Text,Text,NullWritable> {
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, NullWritable> output, Reporter reporter)
				throws IOException {
			int sum = 0;
			String msg = "";
			while (values.hasNext()) {
				String line =values.next().toString();
				String[] strs = line.split("@");
				if(strs.length == 2){
					sum += Integer.parseInt(strs[0]);
					msg += strs[1]+"&";
				}
			}
			msg = key+"|"+sum+"&"+msg;
			output.collect(new Text(msg), null);
		}
	}
	
	
	public static String execute(String tmpIndex) throws IOException{
		System.out.println("begin merge index count");
		JobConf conf = new JobConf();
		conf.setJobName("MergeIndex");
		
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(NullWritable.class);
		
		conf.setMapperClass(Map.class);
		conf.setReducerClass(Reduce.class);
		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		Path in2 = new Path(INDEX_PATH);
		FileSystem system = in2.getFileSystem(conf);
		if(!system.exists(in2)){
			return null;
		}
		
		FileInputFormat.addInputPath(conf, new Path(tmpIndex));//文件
		FileInputFormat.addInputPath(conf, new Path(INDEX_PATH));//目录
		
		
		String tmpFileName = String.valueOf(System.currentTimeMillis());
		Path out = new Path("INDEX_PATH"+tmpFileName);
		system = out.getFileSystem(conf);
		if(system.exists(out)){
			return tmpFileName;
		}
		FileOutputFormat.setOutputPath(conf, out);
		try{
			RunningJob job =JobClient.runJob(conf);
			job.isComplete();
			return tmpFileName;
		}catch(Exception e){
			return null;
		}
	}
}
